package com.zhang.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.zhang.gmall.realtime.utils.DimUtil;
import com.zhang.gmall.realtime.utils.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ExecutorService;

/**
 * @title: 维度数据异步关联
 * @author: zhang
 * @date: 2022/3/8 21:08
 * 模版方法设计模式：在父类中定义完成某个功能的核心算法，具体实现延迟到子类中完成。
 * 在不改变父类核心算法的前提下，每一个子类都有不同的实现
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimAsyncJoinFunction<T> {
    //面向接口编程
    private ExecutorService executorService;
    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        executorService = ThreadPoolUtil.getThreadPoolExecutor();
    }

    //发送异步请求
    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
        //开启多个线程，发送异步请求，完成维度关联
        executorService.submit(
                new Runnable() {
                    @Override
                    public void run() {
                        try {

                            //1.根据流中对象获取维度主键
                            String key = getKey(input);
                            //2.根据维度主键获取维度对象
                            JSONObject dimInfo = DimUtil.getDimInfo(tableName, key);

                            if (dimInfo != null && dimInfo.size() > 0) {
                                //3.将维度对象的属性补充到流中的对象属性上
                                join(input, dimInfo);
                            }

                            resultFuture.complete(Collections.singletonList(input));
                        } catch (Exception e) {
                            e.printStackTrace();
                            throw new RuntimeException("关联维度发生异常～～～");
                        }
                    }
                }
        );
    }

}
